package com.dmp.storm.bolt.kafka;

import com.dmp.storm.bolt.BaseMapBolt;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.support.PropertiesLoaderUtils;
import storm.kafka.bolt.KafkaBolt;
import storm.kafka.bolt.selector.DefaultTopicSelector;
import storm.kafka.bolt.selector.KafkaTopicSelector;

import java.io.IOException;
import java.util.Map;
import java.util.Properties;

public class InsertKafkaBolt extends BaseMapBolt {
	private static final Logger LOG = LoggerFactory
			.getLogger(InsertKafkaBolt.class);

	private static final String DEFAULT_TOPIC = "default";

	private Producer<String, String> producer;
	private KafkaTopicSelector topicSelector;

	// config
	// input
	private String propPath = "kafka.broker.properties";
	private String topic = DEFAULT_TOPIC;
	// output
	private String fields;
	private String[] fieldNames;
	private String separator = "\001";

	@Override
	public void prepare() {
		try {
			Properties properties = PropertiesLoaderUtils
					.loadAllProperties(propPath);
			properties.put(KafkaBolt.TOPIC, topic);

			// for backward compatibility.
			if (topicSelector == null) {
				this.topicSelector = new DefaultTopicSelector(topic);
			}

			ProducerConfig config = new ProducerConfig(properties);
			producer = new Producer<String, String>(config);

			if (StringUtils.isNotBlank(fields)) {
				fieldNames = StringUtils.split(fields, ",");
			}
		} catch (IOException e) {
			LOG.error("failed init InsertKafkaBolt:{}", e.getMessage());
		}
	}

	@Override
	public Map<String, Object> process(Map<String, Object> record) {
		StringBuffer msgBuffer = new StringBuffer();

		if (fieldNames != null && fieldNames.length > 0) {
			if (record != null && record.size() > 0) {
				for (int i = 0; i < fieldNames.length; i++) {
					String fieldName = fieldNames[i];
					Object objValue = record.get(fieldName);
					String value = objValue == null ? "" : objValue.toString();

					msgBuffer.append(value);
					if (i < fieldNames.length - 1) {
						msgBuffer.append(separator);
					}
				}
				// System.out.println(msgBuffer.toString());

				// String[] values = StringUtils.split(msgBuffer.toString(),
				// separator);
				// for(String value : values){
				// System.out.println(value);
				// }
				try {
					producer.send(new KeyedMessage<String, String>(topic,
							msgBuffer.toString()));
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}

		return null;
	}

	public void setPropPath(String propPath) {
		this.propPath = propPath;
	}

	public void setTopic(String topic) {
		if (StringUtils.isNotBlank(topic)) {
			this.topic = topic;
		}
	}

	public void setSeparator(String separator) {
		this.separator = separator;
	}

	public void setFields(String fields) {
		this.fields = fields;
	}
}
